package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"gitee.com/hlwqds/rabitmqTest/config"
	"gitee.com/hlwqds/rabitmqTest/log"
	"gitee.com/hlwqds/rabitmqTest/rabbitmqInAction/concurrenceTask/concurrenceConfig"
	 "gitee.com/hlwqds/rabitmqTest/rabbitmqInAction/concurrenceTask/concurrenceTaskMessage"
	"github.com/streadway/amqp"
)

func addPointsToUser(message []byte){
	meStruct := concurrenceTaskMessage.Message{}
	json.Unmarshal(message, &meStruct)
	fmt.Printf("Adding points to user: %d\n", meStruct.UserId)
}

func addPointsDequeue(ch *amqp.Channel, ex string){
	ch.QueueDeclare(
		concurrenceConfig.ConcurrenceQueueAddPoints,
		false,
		false,
		false,
		false,
		nil,
		)
	ch.QueueBind(
		concurrenceConfig.ConcurrenceQueueAddPoints,
		"",
		ex,
		false,
		nil,
		)

	chAddPoints, err := ch.Consume(
		concurrenceConfig.ConcurrenceQueueAddPoints,
		"addPoints",
		false,
		false,
		false,
		false,
		nil)
	log.FailOnError(err, "ch.Consume err")

	for{
		data, _ := <-chAddPoints
		fmt.Printf("%s\n",data.Body)
		data.Ack(false)
		if bytes.Compare(data.Body, []byte("quit")) == 0 {
			ch.Cancel("addPoints", false)
			return
		} else {
			addPointsToUser(data.Body)
		}
	}
}

func main(){
	conn, err := amqp.Dial(config.GenURLFromConfig())
	log.FailOnError(err, "amqp.Dial error")

	ch, err := conn.Channel()
	log.FailOnError(err, "conn.Channel error")

	ch.ExchangeDeclare(
		concurrenceConfig.ConcurrenceExchange,
		"fanout",
		false,
		false,
		false,
		false,
		nil,
	)

	blockCh := make(chan int)
	go addPointsDequeue(ch, concurrenceConfig.ConcurrenceExchange)
	<-blockCh
}